package com.jensen.metaq.example;

import com.jensen.metaq.example.util.DateUtil;
import com.taobao.metamorphosis.Message;
import com.taobao.metamorphosis.client.producer.MessageProducer;
import com.taobao.metamorphosis.client.producer.SendMessageCallback;
import com.taobao.metamorphosis.client.producer.SendResult;

/**
 * Created by szhua on 2016/6/8.
 */
public class AsyncProducer implements Runnable{

    private MessageProducer producer;
    private String topic="meta-test";
    public static void main(String[] args){
        AsyncProducer asyncProducer=new AsyncProducer();
        Thread thread=new Thread(asyncProducer);
        thread.start();
    }

    public AsyncProducer() {
        super();
        producer=MessageFactory.getMessageProducerInstance();
    }

    public void run() {
        if (producer!=null) {
            System.out.print(producer);
            int count=0;
            producer.publish(topic);
            while (true){
                try {
                    Thread.sleep(5*1000);
                    String msg="current:"+count++;
                    System.out.println(producer+" msg:"+msg);
                    producer.sendMessage(new Message(topic, msg.getBytes()), new SendMessageCallback() {
                        @Override
                        public void onMessageSent(final SendResult result) {
                            if (result.isSuccess()) {
                                System.out.println(DateUtil.getCurrentTime()+" Send message successfully,sent to " + result.getPartition());
                            }
                            else {
                                System.err.println(DateUtil.getCurrentTime()+" Send message failed,error message:" + result.getErrorMessage());
                            }
                        }
                        @Override
                        public void onException(final Throwable e) {
                            e.printStackTrace();
                        }
                    });
                }
                catch (final Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
